package com.lagou.miaosha.rocketmq.mq;

import com.alibaba.fastjson.JSONObject;
import com.lagou.miaosha.rocketmq.mapper.OrderMapper;
import com.lagou.miaosha.rocketmq.pojo.Order;
import com.lagou.miaosha.rocketmq.pojo.Stock;
import com.lagou.miaosha.rocketmq.service.IStockService;
import com.lagou.miaosha.rocketmq.utils.SnowflakeIdWorker;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.redis.util.RedisLockRegistry;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.locks.Lock;

/**
 * @Description TODO
 * @Classname MyConsumeMsgListenerProcessor
 * @Date 2021/8/1 16:25
 * @Author ch'ien
 */

@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "ms", topic = "${TOPIC.MS}")
public class MyConsumeMsgListenerProcessor  implements RocketMQListener<String> {
    
    @Autowired
    private IStockService stockService;

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private RedisLockRegistry redisLockRegistry;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private SnowflakeIdWorker snowflakeIdWorker;


    @Value("${TOPIC.MS_DELAY}")
    private String topicDelay;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(String msg){
        log.info("消息时：{}", msg);
        JSONObject jsonObject = new JSONObject();
        try {
            jsonObject = JSONObject.parseObject(msg);

        } catch (Exception e) {
           log.info("非正常消息");
            return;
        }

        Integer gid = jsonObject.getInteger("gid");
        if (gid == null){
            log.info("没有货物");
            return;
        }

        Lock lock = redisLockRegistry.obtain(String.valueOf(gid));
        try {
            lock.lock();
            Stock stock = stockService.getOneByGid(gid);
            if (stock.getNum()>0){
                //创建订单
                Order order = new Order();
                order.setOrderNo(snowflakeIdWorker.nextId());
                order.setUserId(jsonObject.getInteger("userId"));
                order.setGid(gid);
                order.setStatus(0);
                orderMapper.save(order);

                //30s支付时间
                SendResult sendResult = rocketMQTemplate.syncSend(topicDelay, MessageBuilder.withPayload(order).build(), 3000L, 4);
                if (sendResult != null) {
                    //sendResult不空则表示消息发送成功
                    log.info("send success , send topic = {}, msg = {}, messageId = {}",topicDelay, order, sendResult.getMsgId());
                }
                stockService.decrementStock(stock.getGoodsId());
            }   else {
                stockService.increment(stock.getGoodsId());
            }
        } finally {
            lock.unlock();
        }


    }

}
